/**
 * Software License, Version 1.0
 * 
 * Copyright 2003 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or NaradaBrokering, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called NaradaBrokering, 
 * nor may Indiana University or Community Grids Lab or NaradaBrokering appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 *GENERATED USING SOFTWARE.
 */
package cgl.narada.jms;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;

import javax.jms.JMSException;
import javax.jms.MessageEOFException;
import javax.jms.MessageFormatException;
import javax.jms.MessageNotReadableException;
import javax.jms.MessageNotWriteableException;
import javax.jms.StreamMessage;

import cgl.narada.util.ObjectMarshaller;

/** This class implements the StreamMessage specified in the JMS specification.
    @author Shrideep Pallickara
    $Date$
    $Revision$
*/

public class JmsStreamMessage extends JmsMessage implements StreamMessage {
  private ByteArrayInputStream byteInputStream;
  private ByteArrayOutputStream byteOutputStream;

  private DataInputStream dataInputStream;
  private DataOutputStream dataOutputStream;
  
  private boolean inputStreamsInitialized = false;
  private boolean outputStreamsInitialized = false;
  private byte[] messageBytes;
  private String moduleName = "JmsStreamMessage: ";
  
  public JmsStreamMessage(byte[] _messageBytes) {
    ByteArrayInputStream baInputStream = 
      new ByteArrayInputStream(_messageBytes);
    DataInputStream din = new DataInputStream(baInputStream);
    
    try {
      int headerAndPropertyLength = din.readInt(); 
      byte[] headerPropBytes = new byte[headerAndPropertyLength];
      din.readFully(headerPropBytes);
      super.unmarshallJmsMessage(headerPropBytes);
      
      /** Proceed to unmarshall the bytes */
      messageBytes = new byte[din.readInt()];
      din.readFully(messageBytes);
      initializeOutputStreams();
      initializeInputStreams();
      dataOutputStream.write(messageBytes);
      
    } catch (IOException ioe) {
      System.out.println(moduleName + "Error in unmarshalling");
    }/* end try-catch */
    
    try {
      setJMSType("StreamMessage");
    } catch (JMSException jmsEx) {
      System.out.println(jmsEx);
    }/* end try-catch */
  }


  public JmsStreamMessage() {
    super();
    try {
      setJMSType("StreamMessage");
    } catch (JMSException jmsEx) {
      System.out.println(jmsEx);
    }
  }


  private void initializeInputStreams() {
    byteInputStream = new ByteArrayInputStream(messageBytes);
    dataInputStream = new DataInputStream(byteInputStream);
    inputStreamsInitialized = true;
  }
  
  private void initializeOutputStreams() {
    byteOutputStream = new ByteArrayOutputStream();
    dataOutputStream = new DataOutputStream(byteOutputStream);
    outputStreamsInitialized = true;
  }
  

  /** Puts the message body in read-only mode and repositions the stream
      to the beginning.
      
      @exception JMSException if the JMS provider fails to reset the message
      due to some internal error.
      @exception MessageFormatException if the message has an invalid
      format. */ 
  public void reset() throws JMSException {
    /** This method resets both the messageProperties and messageBody to
	readOnly */
    setBodyAndPropertyAsReadOnly();
    
    try {
      if (dataOutputStream != null) {
	dataOutputStream.flush();
	messageBytes = byteOutputStream.toByteArray();
	byteOutputStream = null;
	dataOutputStream.close();
	dataOutputStream = null;
	outputStreamsInitialized = false;
      } 
      
      if (dataInputStream !=null) {
	byteInputStream = null;
	dataInputStream.close();
	dataInputStream = null;
	inputStreamsInitialized = false;
      }
      
    } catch (IOException ioe) {
      throw new JMSException("JmsBytesMessage" + ioe);
    }
  }

  /** Reads a <code>boolean</code> from the stream message.
   *
   * @return the <code>boolean</code> value read
   *
   * @exception JMSException if the JMS provider fails to read the message
   *                         due to some internal error.
   * @exception MessageEOFException if unexpected end of message stream has
   *                                been reached.     
   * @exception MessageFormatException if this type conversion is invalid.
   * @exception MessageNotReadableException if the message is in write-only 
   *                                        mode.
   */
  
  public boolean readBoolean() throws JMSException {
    /** Do a check to see if we are allowed to read the contents of the
	message body. This message throws a MessageNotReadableException if
	the read is not allowed.*/
    testReadBody();
    initBeforeReads();
    /** Initialize the input streams, this would be done only once per 
	message */
    if (!inputStreamsInitialized) {
      initializeInputStreams();
    }
    boolean returnValue;
    try {
      returnValue = dataInputStream.readBoolean();
    } catch (IOException exception) {
      throw new JMSException("Problems reading the InputStreams " + 
			     exception);
    }
    return returnValue;
  }
  
  
  /** Reads a <code>byte</code> value from the stream message.
   *
   * @return the next byte from the stream message as a 8-bit
   * <code>byte</code>
   *
   * @exception JMSException if the JMS provider fails to read the message
   *                         due to some internal error.
   * @exception MessageEOFException if unexpected end of message stream has
   *                                been reached.     
   * @exception MessageFormatException if this type conversion is invalid.
   * @exception MessageNotReadableException if the message is in write-only 
   *                                        mode.
   */ 
  
  public byte readByte() throws JMSException {
    /** Do a check to see if we are allowed to read the contents of the
	message body. This message throws a MessageNotReadableException if
	the read is not allowed.*/
    testReadBody();
    initBeforeReads();
    /** Initialize the input streams, this would be done only once per 
	message */
    if (!inputStreamsInitialized) {
      initializeInputStreams();
    }
    byte returnValue;
    try {
      returnValue = dataInputStream.readByte();
    } catch (IOException exception) {
      throw new JMSException("Problems reading the InputStreams " + 
			     exception);
    }
    return returnValue;
  }
  
  
  /** Reads a 16-bit integer from the stream message.
   *
   * @return a 16-bit integer from the stream message
   *
   * @exception JMSException if the JMS provider fails to read the message
   *                         due to some internal error.
   * @exception MessageEOFException if unexpected end of message stream has
   *                                been reached.     
   * @exception MessageFormatException if this type conversion is invalid.
   * @exception MessageNotReadableException if the message is in write-only 
   *                                        mode.
   */ 
  
  public short readShort() throws JMSException {
    /** Do a check to see if we are allowed to read the contents of the
	message body. This message throws a MessageNotReadableException if
	the read is not allowed.*/
    testReadBody();
    initBeforeReads();
    /** Initialize the input streams, this would be done only once per 
	message */
    if (!inputStreamsInitialized) {
      initializeInputStreams();
    }
    short returnValue;
    try {
      returnValue = dataInputStream.readShort();
    } catch (IOException exception) {
      throw new JMSException("Problems reading the InputStreams " + 
			     exception);
    }
    return returnValue;
  }
  
  
  /** Reads a Unicode character value from the stream message.
   *
   * @return a Unicode character from the stream message
   *
   * @exception JMSException if the JMS provider fails to read the message
   *                         due to some internal error.
   * @exception MessageEOFException if unexpected end of message stream has
   *                                been reached.     
   * @exception MessageFormatException if this type conversion is invalid      
   * @exception MessageNotReadableException if the message is in write-only 
   *                                        mode.
   */ 
  
  public char readChar() throws JMSException {
      /** Do a check to see if we are allowed to read the contents of the
	message body. This message throws a MessageNotReadableException if
	the read is not allowed.*/
    testReadBody();
    initBeforeReads();
    /** Initialize the input streams, this would be done only once per 
	message */
    if (!inputStreamsInitialized) {
      initializeInputStreams();
    }
    char returnValue;
    try {
      returnValue = dataInputStream.readChar();
    } catch (IOException exception) {
      throw new JMSException("Problems reading the InputStreams " + 
			     exception);
    }
    return returnValue;
  }
  
  
  /** Reads a 32-bit integer from the stream message.
   *
   * @return a 32-bit integer value from the stream message, interpreted
   * as an <code>int</code>
   *
   * @exception JMSException if the JMS provider fails to read the message
   *                         due to some internal error.
   * @exception MessageEOFException if unexpected end of message stream has
   *                                been reached.     
   * @exception MessageFormatException if this type conversion is invalid.
   * @exception MessageNotReadableException if the message is in write-only 
   *                                        mode.
   */ 
  
  public int readInt() throws JMSException {
      /** Do a check to see if we are allowed to read the contents of the
	message body. This message throws a MessageNotReadableException if
	the read is not allowed.*/
    testReadBody();
    initBeforeReads();
    /** Initialize the input streams, this would be done only once per 
	message */
    if (!inputStreamsInitialized) {
      initializeInputStreams();
    }
    int returnValue;
    try {
      returnValue = dataInputStream.readInt();
    } catch (IOException exception) {
      throw new JMSException("Problems reading the InputStreams " + 
			     exception);
    }
    return returnValue;
  }
  
  
  /** Reads a 64-bit integer from the stream message.
   *
   * @return a 64-bit integer value from the stream message, interpreted as
   * a <code>long</code>
   *
   * @exception JMSException if the JMS provider fails to read the message
   *                         due to some internal error.
   * @exception MessageEOFException if unexpected end of message stream has
   *                                been reached.     
   * @exception MessageFormatException if this type conversion is invalid.
   * @exception MessageNotReadableException if the message is in write-only 
   *                                        mode.
   */ 
  
  public long readLong() throws JMSException {
    /** Do a check to see if we are allowed to read the contents of the
	message body. This message throws a MessageNotReadableException if
	the read is not allowed.*/
    testReadBody();
    initBeforeReads();
    /** Initialize the input streams, this would be done only once per 
	message */
    if (!inputStreamsInitialized) {
      initializeInputStreams();
    }
    long returnValue;
    try {
      returnValue = dataInputStream.readLong();
    } catch (IOException exception) {
      throw new JMSException("Problems reading the InputStreams " + 
			     exception);
    }
    return returnValue;
  }
  
  
  /** Reads a <code>float</code> from the stream message.
   *
   * @return a <code>float</code> value from the stream message
   *
   * @exception JMSException if the JMS provider fails to read the message
   *                         due to some internal error.
   * @exception MessageEOFException if unexpected end of message stream has
   *                                been reached.     
   * @exception MessageFormatException if this type conversion is invalid.
   * @exception MessageNotReadableException if the message is in write-only 
   *                                        mode.
   */ 
  
  public float readFloat() throws JMSException {
    /** Do a check to see if we are allowed to read the contents of the
	message body. This message throws a MessageNotReadableException if
	the read is not allowed.*/
    testReadBody();
    initBeforeReads();
    /** Initialize the input streams, this would be done only once per 
	message */
    if (!inputStreamsInitialized) {
      initializeInputStreams();
    }
    float returnValue;
    try {
      returnValue = dataInputStream.readFloat();
    } catch (IOException exception) {
      throw new JMSException("Problems reading the InputStreams " + 
			     exception);
    }
    return returnValue;
  }
  
  
  /** Reads a <code>double</code> from the stream message.
   *
   * @return a <code>double</code> value from the stream message
   *
   * @exception JMSException if the JMS provider fails to read the message
   *                         due to some internal error.
   * @exception MessageEOFException if unexpected end of message stream has
   *                                been reached.     
   * @exception MessageFormatException if this type conversion is invalid.
   * @exception MessageNotReadableException if the message is in write-only 
   *                                        mode.
   */ 
  
  public double readDouble() throws JMSException {
    /** Do a check to see if we are allowed to read the contents of the
	message body. This message throws a MessageNotReadableException if
	the read is not allowed.*/
    testReadBody();
    initBeforeReads();
    /** Initialize the input streams, this would be done only once per 
	message */
    if (!inputStreamsInitialized) {
      initializeInputStreams();
    }
    double returnValue;
    try {
      returnValue = dataInputStream.readDouble();
    } catch (IOException exception) {
      throw new JMSException("Problems reading the InputStreams " + 
			     exception);
    }
    return returnValue;
    
  }
  
  
  /** Reads a <CODE>String</CODE> from the stream message.
   *
   * @return a Unicode string from the stream message
   *
   * @exception JMSException if the JMS provider fails to read the message
   *                         due to some internal error.
   * @exception MessageEOFException if unexpected end of message stream has
   *                                been reached.     
   * @exception MessageFormatException if this type conversion is invalid.
   * @exception MessageNotReadableException if the message is in write-only 
   *                                        mode.
   */ 
  
  public String readString() throws JMSException {
      /** Do a check to see if we are allowed to read the contents of the
	message body. This message throws a MessageNotReadableException if
	the read is not allowed.*/
    testReadBody();
    initBeforeReads();
    /** Initialize the input streams, this would be done only once per 
	message */
    if (!inputStreamsInitialized) {
      initializeInputStreams();
    }
    String returnValue;
    try {
      returnValue = dataInputStream.readUTF();
    } catch (IOException exception) {
      throw new JMSException("Problems reading the InputStreams " + 
			     exception);
    }
    return returnValue;
  }
  
  
  /** Reads a byte array field from the stream message into the 
   * specified <CODE>byte[]</CODE> object (the read buffer). 
   * 
   * <P>To read the field value, <CODE>readBytes</CODE> should be 
   * successively called 
   * until it returns a value less than the length of the read buffer.
   * The value of the bytes in the buffer following the last byte 
   * read is undefined.
   * 
   * <P>If <CODE>readBytes</CODE> returns a value equal to the length of the 
   * buffer, a subsequent <CODE>readBytes</CODE> call must be made. If there 
   * are no more bytes to be read, this call returns -1.
   * 
   * <P>If the byte array field value is null, <CODE>readBytes</CODE> 
   * returns -1.
   *
   * <P>If the byte array field value is empty, <CODE>readBytes</CODE> 
   * returns 0.
   * 
   * <P>Once the first <CODE>readBytes</CODE> call on a <CODE>byte[]</CODE>
   * field value has been made,
   * the full value of the field must be read before it is valid to read 
   * the next field. An attempt to read the next field before that has 
   * been done will throw a <CODE>MessageFormatException</CODE>.
   * 
   * <P>To read the byte field value into a new <CODE>byte[]</CODE> object, 
   * use the <CODE>readObject</CODE> method.
   *
   * @param value the buffer into which the data is read
   *
   * @return the total number of bytes read into the buffer, or -1 if 
   * there is no more data because the end of the byte field has been 
   * reached
   *
   * @exception JMSException if the JMS provider fails to read the message
   *                         due to some internal error.
   * @exception MessageEOFException if unexpected end of message stream has
   *                                been reached.     
   * @exception MessageFormatException if this type conversion is invalid.
   * @exception MessageNotReadableException if the message is in write-only 
   *                                        mode.
   * 
   * @see #readObject()
   */ 
  
  public int readBytes(byte[] value) throws JMSException {
    /** Do a check to see if we are allowed to read the contents of the
	message body. This message throws a MessageNotReadableException if
	the read is not allowed.*/
    int length = value.length;
    testReadBody();
    initBeforeReads();
    /** Initialize the input streams, this would be done only once per 
	message */
    if (!inputStreamsInitialized) {
      initializeInputStreams();
    }
    int bytesRead=-1;
    if (length < 0 || length > value.length) {
      throw new JMSException("length < 0 or > the array size");
    }

    try {
      dataInputStream.mark(length);
      int available = dataInputStream.available();
      if (available == 0) {
	/** The end of stream was reached. We are required to return -1.*/
	bytesRead= -1;
      } else if (value.length <= available) {
	bytesRead = value.length;
	dataInputStream.readFully(value);
      } else {
	dataInputStream.readFully(value, 0, available);
	bytesRead = available;
      }
      
    } catch (IOException exception) {
      throw new JMSException("Problems reading the InputStreams " + 
			     exception);
    }
    return bytesRead;
  }
  
  
  /** Reads an object from the stream message.
   *
   * <P>This method can be used to return, in objectified format,
   * an object in the Java programming language ("Java object") that has 
   * been written to the stream with the equivalent
   * <CODE>writeObject</CODE> method call, or its equivalent primitive
   * <CODE>write<I>type</I></CODE> method.
   *  
   * <P>Note that byte values are returned as <CODE>byte[]</CODE>, not 
   * <CODE>Byte[]</CODE>.
   *
   * <P>An attempt to call <CODE>readObject</CODE> to read a byte field 
   * value into a new <CODE>byte[]</CODE> object before the full value of the
   * byte field has been read will throw a 
   * <CODE>MessageFormatException</CODE>.
   *
   * @return a Java object from the stream message, in objectified
   * format (for example, if the object was written as an <CODE>int</CODE>, 
   * an <CODE>Integer</CODE> is returned)
   *
   * @exception JMSException if the JMS provider fails to read the message
   *                         due to some internal error.
   * @exception MessageEOFException if unexpected end of message stream has
   *                                been reached.     
   * @exception MessageFormatException if this type conversion is invalid.
   * @exception MessageNotReadableException if the message is in write-only 
   *                                        mode.
   * 
   * @see #readBytes(byte[] value)
   */ 

  public Object readObject() throws JMSException {
    /** Do a check to see if we are allowed to read the contents of the
	message body. This message throws a MessageNotReadableException if
	the read is not allowed.*/
    testReadBody();
    initBeforeReads();
    /** Initialize the input streams, this would be done only once per 
	message */
    if (!inputStreamsInitialized) {
      initializeInputStreams();
    }

    Object returnObject;

    try {
      int length = dataInputStream.readInt();
      byte[] objectBytes = new byte[length];
      dataInputStream.read(objectBytes);
      returnObject = ObjectMarshaller.unmarshall(objectBytes);
      
    } catch (IOException exception) {
      throw new JMSException("Problems reading the InputStreams " + 
			     exception);
    } catch (ClassNotFoundException cnfe) {
      throw new JMSException("Problems reading the InputStreams " + 
			     cnfe);
    }
    return returnObject;
  }
  
  

  /** Writes a <code>boolean</code> to the stream message.
   * The value <code>true</code> is written as the value 
   * <code>(byte)1</code>; the value <code>false</code> is written as 
   * the value <code>(byte)0</code>.
   *
   * @param value the <code>boolean</code> value to be written
   *
   * @exception JMSException if the JMS provider fails to write the message
   *                         due to some internal error.
   * @exception MessageNotWriteableException if the message is in read-only 
   *                                         mode.
   */
  
  public void writeBoolean(boolean value) throws JMSException {
    /** Do a check to see if we are allowed to make changes to the body of 
	the message. This method throws a MessageNotWritableException if the
	write is not allowed */
    testWriteBody();
    /** Initialize streams, this would be done only once. */
    if (!outputStreamsInitialized) {
      initializeOutputStreams();
    }
    
    try {
      dataOutputStream.writeBoolean(value);
    } catch (IOException exception) {
      throw new JMSException("Problems writing to OutputStreams " + 
			     exception);
    } 
  }

  /** Writes a <code>byte</code> to the stream message.
   *
   * @param value the <code>byte</code> value to be written
   *
   * @exception JMSException if the JMS provider fails to write the message
   *                         due to some internal error.
   * @exception MessageNotWriteableException if the message is in read-only 
   *                                         mode.
   */ 
  
  public  void  writeByte(byte value) throws JMSException {
    /** Do a check to see if we are allowed to make changes to the body of 
	the message. This method throws a MessageNotWritableException if the
	write is not allowed */
    testWriteBody();
    /** Initialize streams, this would be done only once. */
    if (!outputStreamsInitialized) {
      initializeOutputStreams();
    }
    
    try {
      dataOutputStream.writeByte(value);
    } catch (IOException exception) {
      throw new JMSException("Problems writing to OutputStreams " + 
			     exception);
    } 
  }
  
  
  /** Writes a <code>short</code> to the stream message.
   *
   * @param value the <code>short</code> value to be written
   *
   * @exception JMSException if the JMS provider fails to write the message
   *                         due to some internal error.
   * @exception MessageNotWriteableException if the message is in read-only 
   *                                         mode.
   */ 
  
  public void  writeShort(short value) throws JMSException {
    /** Do a check to see if we are allowed to make changes to the body of 
	the message. This method throws a MessageNotWritableException if the
	write is not allowed */
    testWriteBody();
    /** Initialize streams, this would be done only once. */
    if (!outputStreamsInitialized) {
      initializeOutputStreams();
    }
    
    try {
      dataOutputStream.writeShort(value);
    } catch (IOException exception) {
      throw new JMSException("Problems writing to OutputStreams " + 
			     exception);
    }
  }
  
  
  /** Writes a <code>char</code> to the stream message.
   *
   * @param value the <code>char</code> value to be written
   *
   * @exception JMSException if the JMS provider fails to write the message
   *                         due to some internal error.
   * @exception MessageNotWriteableException if the message is in read-only 
   *                                         mode.
   */ 
  
    public void writeChar(char value) throws JMSException {
     /** Do a check to see if we are allowed to make changes to the body of 
	the message. This method throws a MessageNotWritableException if the
	write is not allowed */
    testWriteBody();
    /** Initialize streams, this would be done only once. */
    if (!outputStreamsInitialized) {
      initializeOutputStreams();
    }
    
    try {
      dataOutputStream.writeChar(value);
    } catch (IOException exception) {
      throw new JMSException("Problems writing to OutputStreams " + 
			     exception);
    } 
  }
  
  
  /** Writes an <code>int</code> to the stream message.
   *
   * @param value the <code>int</code> value to be written
   *
   * @exception JMSException if the JMS provider fails to write the message
   *                         due to some internal error.
   * @exception MessageNotWriteableException if the message is in read-only 
   *                                         mode.
   */ 
  
  public void writeInt(int value) throws JMSException {
     /** Do a check to see if we are allowed to make changes to the body of 
	the message. This method throws a MessageNotWritableException if the
	write is not allowed */
    testWriteBody();
    /** Initialize streams, this would be done only once. */
    if (!outputStreamsInitialized) {
      initializeOutputStreams();
    }
    
    try {
      dataOutputStream.writeInt(value);
    } catch (IOException exception) {
      throw new JMSException("Problems writing to OutputStreams " + 
			     exception);
    } 
  }
  
  
  /** Writes a <code>long</code> to the stream message.
   *
   * @param value the <code>long</code> value to be written
   *
   * @exception JMSException if the JMS provider fails to write the message
   *                         due to some internal error.
   * @exception MessageNotWriteableException if the message is in read-only 
   *                                         mode.
   */ 
  
  public void  writeLong(long value) throws JMSException {
     /** Do a check to see if we are allowed to make changes to the body of 
	the message. This method throws a MessageNotWritableException if the
	write is not allowed */
    testWriteBody();
    /** Initialize streams, this would be done only once. */
    if (!outputStreamsInitialized) {
      initializeOutputStreams();
    }
    
    try {
      dataOutputStream.writeLong(value);
    } catch (IOException exception) {
      throw new JMSException("Problems writing to OutputStreams " + 
			     exception);
    } 
  }
  
  
  /** Writes a <code>float</code> to the stream message.
   *
   * @param value the <code>float</code> value to be written
   *
   * @exception JMSException if the JMS provider fails to write the message
   *                         due to some internal error.
   * @exception MessageNotWriteableException if the message is in read-only 
   *                                         mode.
   */ 
  
  public void writeFloat(float value) throws JMSException {
     /** Do a check to see if we are allowed to make changes to the body of 
	the message. This method throws a MessageNotWritableException if the
	write is not allowed */
    testWriteBody();
    /** Initialize streams, this would be done only once. */
    if (!outputStreamsInitialized) {
      initializeOutputStreams();
    }
    
    try {
      dataOutputStream.writeFloat(value);
    } catch (IOException exception) {
      throw new JMSException("Problems writing to OutputStreams " + 
			     exception);
    } 
  }
  
  
  /** Writes a <code>double</code> to the stream message.
   *
   * @param value the <code>double</code> value to be written
   *
   * @exception JMSException if the JMS provider fails to write the message
   *                         due to some internal error.
   * @exception MessageNotWriteableException if the message is in read-only 
   *                                         mode.
   */ 
  
  public void 
  writeDouble(double value) throws JMSException {
     /** Do a check to see if we are allowed to make changes to the body of 
	the message. This method throws a MessageNotWritableException if the
	write is not allowed */
    testWriteBody();
    /** Initialize streams, this would be done only once. */
    if (!outputStreamsInitialized) {
      initializeOutputStreams();
    }
    
    try {
      dataOutputStream.writeDouble(value);
    } catch (IOException exception) {
      throw new JMSException("Problems writing to OutputStreams " + 
			     exception);
    } 
    
  }
  
  
  /** Writes a <code>String</code> to the stream message.
   *
   * @param value the <code>String</code> value to be written
   *
   * @exception JMSException if the JMS provider fails to write the message
   *                         due to some internal error.
   * @exception MessageNotWriteableException if the message is in read-only 
   *                                         mode.
   */ 
  
  public void writeString(String value) throws JMSException {
     /** Do a check to see if we are allowed to make changes to the body of 
	the message. This method throws a MessageNotWritableException if the
	write is not allowed */
    testWriteBody();
    /** Initialize streams, this would be done only once. */
    if (!outputStreamsInitialized) {
      initializeOutputStreams();
    }
    
    try {
      dataOutputStream.writeUTF(value);
    } catch (IOException exception) {
      throw new JMSException("Problems writing to OutputStreams " + 
			     exception);
    }

  }
  
  
  /** Writes a byte array field to the stream message.
   *
   * <P>The byte array <code>value</code> is written to the message
   * as a byte array field. Consecutively written byte array fields are 
   * treated as two distinct fields when the fields are read.
   * 
   * @param value the byte array value to be written
   *
   * @exception JMSException if the JMS provider fails to write the message
   *                         due to some internal error.
   * @exception MessageNotWriteableException if the message is in read-only 
   *                                         mode.
   */ 
  
  public void
  writeBytes(byte[] value) throws JMSException {
    /** Do a check to see if we are allowed to make changes to the body of 
	the message. This method throws a MessageNotWritableException if the
	write is not allowed */
    testWriteBody();
    /** Initialize streams, this would be done only once. */
    if (!outputStreamsInitialized) {
      initializeOutputStreams();
    }
    
    try {
      dataOutputStream.write(value);
    } catch (IOException exception) {
      throw new JMSException("Problems writing to OutputStreams " + 
			     exception);
    } 
    
  }
  
  
  /** Writes a portion of a byte array as a byte array field to the stream 
   * message.
   *  
   * <P>The a portion of the byte array <code>value</code> is written to the
   * message as a byte array field. Consecutively written byte 
   * array fields are treated as two distinct fields when the fields are 
   * read.
   *
   * @param value the byte array value to be written
   * @param offset the initial offset within the byte array
   * @param length the number of bytes to use
   *
   * @exception JMSException if the JMS provider fails to write the message
   *                         due to some internal error.
   * @exception MessageNotWriteableException if the message is in read-only 
   *                                         mode.
   */ 
  
  public void  writeBytes(byte[] value, int offset, int length) 
    throws JMSException {
    /** Do a check to see if we are allowed to make changes to the body of 
	the message. This method throws a MessageNotWritableException if the
	write is not allowed */
    testWriteBody();
    /** Initialize streams, this would be done only once. */
    if (!outputStreamsInitialized) {
      initializeOutputStreams();
    }
    
    try {
      dataOutputStream.write(value, offset, length);
    } catch (IOException exception) {
      throw new JMSException("Problems writing to OutputStreams " + 
			     exception);
    } 
  }
  
  
  /** Writes an object to the stream message.
   *
   * <P>This method works only for the objectified primitive
   * object types (<code>Integer</code>, <code>Double</code>, 
   * <code>Long</code>&nbsp;...), <code>String</code> objects, and byte 
   * arrays.
   *
   * @param value the Java object to be written
   *
   * @exception JMSException if the JMS provider fails to write the message
   *                         due to some internal error.
   * @exception MessageFormatException if the object is invalid.
   * @exception MessageNotWriteableException if the message is in read-only 
   *                                         mode.
   */ 
  
  public void writeObject(Object value) throws JMSException {
    if (value == null) {
      throw new JMSException("Specified value is null");
    }
     /** Do a check to see if we are allowed to make changes to the body of 
	the message. This method throws a MessageNotWritableException if the
	write is not allowed */
    testWriteBody();
    /** Initialize streams, this would be done only once. */
    if (!outputStreamsInitialized) {
      initializeOutputStreams();
    }
    
    try {
      byte[] objectBytes = ObjectMarshaller.marshall(value);
      dataOutputStream.write(objectBytes.length);
      dataOutputStream.write(objectBytes);
    } catch (IOException exception) {
      throw new JMSException("Problems writing to OutputStreams " + 
			     exception);
    }     
  }

  
  public byte[] getBytes() {
    ByteArrayOutputStream baOutputStream = new ByteArrayOutputStream();
    DataOutputStream dout = new DataOutputStream(baOutputStream);
    byte[] marshalledBytes = null;
    try {
      byte[] headerPropBytes = super.getBytes();
      /* Write the length of the byte Stream comprising headers and properties
	 first.*/
      dout.writeInt(headerPropBytes.length);
      dout.write(headerPropBytes);
      
      /** Proceed to marshall the byte array */
      dataOutputStream.flush();
      messageBytes = byteOutputStream.toByteArray();
      dout.writeInt(messageBytes.length);
      dout.write(messageBytes);
      
      dout.flush();
      marshalledBytes = baOutputStream.toByteArray();
    } catch (IOException ioe) {
      System.out.println(moduleName + "Error marshalling" + ioe);
    }/* end try-catch block */
    return marshalledBytes;
  }
  
  
  private void initBeforeReads() throws JMSException{
    if (dataInputStream == null) {
      initializeInputStreams();
    }
    try {
      dataInputStream.mark(messageBytes.length - dataInputStream.available());
    } catch(IOException ioe) {
      throw new JMSException("JmsBytesMessage" + ioe);
    }
  }

  public void clearBody() throws JMSException {
    initializeOutputStreams();
  }
}
